package MapReduce.Demo12_BookJoin.reduceJoin;

import MapReduce.writableBean.BookBorrowWritable;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import utils.JobSubmit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author:lixinlei
 * @Date:2022/3/24 9:20
 **/
public class BookReduceJoinSecondApp {

    public static class BRJSMapper extends Mapper<LongWritable, Text, BookBorrowWritable, NullWritable> {

        BookBorrowWritable outKey = new BookBorrowWritable();
        NullWritable outValue = NullWritable.get();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();

            String[] fields = line.split("\t");

            //如果字段数量为2，表示此条数据为人员数据
            if(null!=fields && fields.length==2){
                outKey.setPersonId(fields[0]);
                outKey.setPersonName(fields[1]);
                outKey.setFlag("person");
            }else if(null!=fields && fields.length==5){
                //如果字段数量为4，表示此条数据为第一步合并后的数据
                outKey.setBookId(fields[0]);
                outKey.setBookName(fields[1]);
                outKey.setPersonId(fields[2]);
                outKey.setDateTime(fields[4]);
                outKey.setFlag("join");
            }

            context.write(outKey,outValue);

        }
    }

    public static class BRJRSeducer extends Reducer<BookBorrowWritable,NullWritable,BookBorrowWritable,NullWritable> {

        @Override
        protected void reduce(BookBorrowWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

            //接收每个相同人员id下的人员姓名
            String personName = "";
            //初始化一个第一步合并后的记录类型数据的集合
            List<BookBorrowWritable> borrowList = new ArrayList<BookBorrowWritable>();

            //遍历所有的value，但是获取的是每个value所对应的key的内容
            for (NullWritable value : values) {
                //获取同组数据下的每条数据的标识
                String flag = key.getFlag();
                //如果是第一步合并后的记录数据，先把所有的数据封装在一个集合里，后续再遍历拼接人员姓名
                if(flag.equals("join")){
                    try {
                        BookBorrowWritable borrowInfo = new BookBorrowWritable();
                        //阿里不建议使用
                        BeanUtils.copyProperties(borrowInfo,key);
                        borrowList.add(borrowInfo);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }else if(flag.equals("person")){

                    //如果是人员数据 提取人员姓名
                    personName = key.getPersonName();
                }
            }

            //在已经获取到图书名的前提下，把所有的借阅数据再次遍历，赋图书名字段值
            for (BookBorrowWritable borrowInfo : borrowList) {
                borrowInfo.setPersonName(personName);
                System.out.println("========"+borrowInfo);
                context.write(borrowInfo,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(
                BookReduceJoinSecondApp.class,
                args,
                BookReduceJoinSecondGroupComparator.class);
    }

}
